package skewreduce.framework.planner.partition;

import org.apache.hadoop.util.ReflectionUtils;

import skewreduce.framework.planner.PartitionNode;
import skewreduce.framework.planner.axis.Axis;
import skewreduce.framework.planner.axis.AxisChooser;
import skewreduce.framework.planner.axis.AxisChooser.PreparedAxes;


public class CompositePartitioner extends Partitioner {
	AxisChooser axisChooser;
	Axis[] axesBuf = new Axis[1];
	
	@Override
	protected void setup() {
		Class<? extends AxisChooser> cls = conf.getClass("skewreduce.planner.partitioner.axisChooser", null, AxisChooser.class);
		axisChooser = ReflectionUtils.newInstance(cls, conf);
	}
	
	@Override
	public PartitionNode[] partition(PartitionNode root) {
		PreparedAxes axes = AxisChooser.prepareAxes(getModel(), root);
		if ( axes == null ) return null;
		
		Axis axis = axisChooser.choose(root,axes.getAxes());
		axesBuf[0] = axis;
		return doPartition(root,axesBuf,axes.getSortOrder());
	}
}
